package com.shujia.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code02SparkSubmit extends App {


  /**
   * 提交命令：
   * spark-submit --class  com.shujia.core.Code02SparkSubmit --executor-memory 512m --total-executor-cores 1 spark_code-1.0.jar
   * spark-submit  --class   com.shujia.core.Code02SparkSubmit --master yarn --executor-memory 512m --total-executor-cores 1 spark_code-1.0.jar
   */


  // 1.如果要使用Spark进行数据计算，那么需要先设置Spark的执行环境
  private val conf: SparkConf = new SparkConf()
  conf.setAppName("WordCount")
  //  conf.setMaster("yarn") // 设置运行模式为yarn

  private val sc = new SparkContext(conf) // 获取操作类的对象 -> 通常称为 sc

  // 2.获取操作对象后，再读取数据
  //   读取数据后返回的是一个RDD其中 数据格式泛型为String 类比 Scala中的List[String] 注意：RDD不是一个容器，并不会保存数据
  //   minPartitions 最小分区数据表示限定下限 那么对于切片数如果和最小分区数不匹配，那么会以倍数形式增加
  private val wordFileData: RDD[String] = sc.textFile("/input/words.txt", 4)
  println("wordFileDataRDDPartition:" + wordFileData.getNumPartitions)

  // 3.数据计算
  private val flatMapWord: RDD[String] = wordFileData
    //    .foreach(x => println("line:" + x))
    .flatMap(
      (line: String) => {
        // java,spark,java,hadoop
        println("flatmap执行了...")
        line.split(",")
      }

    )
  println("flatMapWordRDDPartition:" + flatMapWord.getNumPartitions)


  private val firstMap: RDD[String] = flatMapWord.map {
    //  [java,spark,java,hadoop] => 每个单词都会触发一次map计算逻辑
    case word => {
      println("first map running...")
      word
    }
  }


  private val groupByWord: RDD[(String, Iterable[String])] = firstMap.groupBy(x => x, 2)

  println("groupByWordRDDPartition:" + groupByWord.getNumPartitions)


  private val mapRes: RDD[(String, Int)] = groupByWord.map {
    case (word, wordLine) => {
      println("map执行了...")
      (word, wordLine.size)
    }
  }

  println("mapResRDDPartition:" + mapRes.getNumPartitions)

  // Job是由特定函数去触发的
  //  mapRes.foreach(println) // 通过两个foreach函数分别触发了两个job任务
  //  mapRes.foreach(println)

  var path: String = "/output/spark/wordCount"

  private val configuration: Configuration = sc.hadoopConfiguration

  private val fileSystem: FileSystem = FileSystem.get(configuration)

  if (fileSystem.exists(new Path(path))) {
    println(s"path:$path 路径已存在，开始删除..")
    fileSystem.delete(new Path(path), true)
  }

  // HDFS当中
  mapRes.saveAsTextFile(path)


}
